package xdb;

import com.goldhuman.Common.Marshal.OctetsStream;

import java.util.Iterator;
import java.util.concurrent.ConcurrentMap;

public class CheckpointCache extends ThreadHelper {

    static int PROCESS_TIME = 1000; //处理时间间隔为1000ms
    static int PROCESS_MAX_COUNT = 1000; //每个存储周期每张数据表一次存储的数据量为1000条
    static int PROCESS_BATCH = 10; // 小于10条单次处理，大于等于10条批量处理
    static int PROCESS_SINGLE = 100; // 单次处理的最大数量为100条
    private final ConcurrentMap<String, CheckpointCacheInfo> cacheTables = xdb.util.Misc.newConcurrentMap();

    public CheckpointCache() {
        super("xdb.CheckpointCache");
    }

    public static void setProcessTime(String strtm) {
        try {
            PROCESS_TIME = Integer.valueOf(strtm);
            xdb.Trace.info("PROCESS_TIME is " + PROCESS_TIME);
        } catch (Exception e) {
            xdb.Trace.error("strtm = " + strtm, e);
        }
    }

    public static void setProcessMaxCount(String strcount) {
        try {
            PROCESS_MAX_COUNT = Integer.valueOf(strcount);
            xdb.Trace.info("PROCESS_MAX_COUNT is " + PROCESS_MAX_COUNT);
        } catch (Exception e) {
            xdb.Trace.error("strcount = " + strcount, e);
        }
    }

    public static void setProcessBatch(String strbatch) {
        try {
            PROCESS_BATCH = Integer.valueOf(strbatch);
            xdb.Trace.info("PROCESS_BATCH is " + PROCESS_BATCH);
        } catch (Exception e) {
            xdb.Trace.error("strbatch = " + strbatch, e);
        }
    }

    public static void setProcessSingle(String strsingle) {
        try {
            PROCESS_SINGLE = Integer.valueOf(strsingle);
            xdb.Trace.info("PROCESS_SINGLE is " + PROCESS_SINGLE);
        } catch (Exception e) {
            xdb.Trace.error("strsingle = " + strsingle, e);
        }
    }

    @Override
    public void run() {
        while (super.isRunning()) {
            long now = System.currentTimeMillis();
            this.save(PROCESS_MAX_COUNT);
            long useTime = System.currentTimeMillis() - now;
            if (useTime >= PROCESS_TIME) {
                super.sleepIdle(100);
                xdb.Trace.info("process time is " + useTime + "ms!");
            } else {
                super.sleepIdle(PROCESS_TIME - useTime);
            }
        }

        Trace.warn("final CheckPointCache begin");
        this.save(Integer.MAX_VALUE);
        Trace.warn("final CheckPointCache end");

        //关服最后一次存储，多尝试几次，避免数据存储不上导致数据不一致
        int totalCacheCount = totalCacheCount();
        if (totalCacheCount > 0) {
            Trace.error("final CheckPointCache left " + totalCacheCount);
            for (int i = 0; i < 60; i++) {
                this.save(Integer.MAX_VALUE);
                totalCacheCount = totalCacheCount();
                if (totalCacheCount == 0) {
                    break;
                } else {
                    Trace.error("final CheckPointCache i=" + i + " left=" + totalCacheCount);
                }
                super.sleepIdle(10000);
            }
        }
    }

    private int totalCacheCount() {
        int totalCacheCount = 0;
        Iterator<CheckpointCacheInfo> iter = cacheTables.values().iterator();
        while (iter.hasNext()) {
            totalCacheCount += iter.next().cacheCount();
        }
        return totalCacheCount;
    }

    private void save(int maxCount) {
        Iterator<CheckpointCacheInfo> iter = cacheTables.values().iterator();
        while (iter.hasNext()) {
            try {
                iter.next().save(maxCount);
            } catch (Exception e) {
                //save时可能异常，需捕获，否则线程可能停止工作
                Trace.error("CheckpointCache save error", e);
            }
        }
    }

    public void flushRecord(String tableName, String key, xdb.TRecord.State state, OctetsStream value, long flushTime) {
        if (cacheTables.containsKey(tableName) == false) {
            cacheTables.put(tableName, new CheckpointCacheInfo(tableName));
        }
        cacheTables.get(tableName).addRecord(key, state, value, flushTime);
    }

    public boolean existInCache(String tableName, String key) {
        if (!cacheTables.containsKey(tableName)) {
            return false;
        }
        return cacheTables.get(tableName).existInCache(key);
    }

    public OctetsStream find(String tableName, String key) {
        if (!cacheTables.containsKey(tableName)) {
            return null;
        }
        return cacheTables.get(tableName).find(key);
    }
}
